package com.example.demo.thread.juc;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Description: 阻塞队列实现的生产者 - 消费者模型
 * @Author: LLL
 * @Date: 2020-03-20 13:21
 **/
@Slf4j
class MyResourse {
    private volatile boolean flag = true;

    private AtomicInteger atomicInteger = new AtomicInteger();

    BlockingQueue<String> blockingQueue = null;

    public MyResourse(BlockingQueue<String> blockingQueue) {
        log.info("blockingQueue typ is :[{}]", blockingQueue.getClass().getName());
        this.blockingQueue = blockingQueue;
    }

    @SneakyThrows public void myProd() {
        String data = "";
        boolean ret;
        while (flag) {
            data = "" + atomicInteger.incrementAndGet();// ++i的意思
            ret = blockingQueue.offer(data, 2, TimeUnit.SECONDS);// 不报异常
            if (ret) {
                log.info("{}插入了数据：{}", Thread.currentThread().getName(), data);
            } else {
                log.info("{}插入数据：{}失败！", Thread.currentThread().getName(), data);
            }
            TimeUnit.SECONDS.sleep(1);
        }
        log.info("不需要生产");
    }

    @SneakyThrows public void myConsume() {
        String res = null;
        while (flag) {
            res = blockingQueue.poll(2, TimeUnit.SECONDS);
            log.info("{}消费了了数据：{}", Thread.currentThread().getName(), res);
            if (StringUtils.isEmpty(res)) {
                flag = false;
                log.info("{}超过两秒没有取到，退出", Thread.currentThread().getName());
                return;
            }
        }
    }

    public void stop() {
        this.flag = false;
    }

}

@Slf4j
public class ProdConsumer_BlockQueueDemo {
    @SneakyThrows public static void main(String[] args) {
        MyResourse resourse = new MyResourse(new ArrayBlockingQueue<>(3));
        new Thread(() -> {
            log.info("生产线程启动......");
            resourse.myProd();
        }, "prod").start();

        new Thread(() -> {
            log.info("消费线程启动......");
            resourse.myConsume();
        }, "consume").start();
        Thread.sleep(5000);
        log.info("正在关闭.....");
        resourse.stop();
    }
}
